package day08

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 第2章	Spark SQL编程
 *
 * 本章重点学习如何使用 DataFrame和DataSet进行编程，以及他们之间的关系和转换，关于具体的SQL书写不是本章的重点。
 *
 * 2.1	SparkSession新的起始点
 * 在老的版本中，SparkSQL提供两种SQL查询起始点：一个叫SQLContext，用于Spark自己提供的SQL查询；
 * 一个叫HiveContext，用于连接Hive的查询。
 *
 * SparkSession是Spark最新的SQL查询起始点，实质上是 SQLContext 和 HiveContext 的组合，所以在 SQLContext 和
 * HiveContext上可用的API在SparkSession上同样是可以使用的。SparkSession内部封装了sparkContext，
 * 所以计算实际上是由sparkContext完成的。当我们使用 spark-shell 的时候, spark 会自动的创建一个叫做
 * spark的SparkSession, 就像我们以前可以自动获取到一个sc来表示SparkContext
 *
 * 2.2	DataFrame（通过spark-local模式，进入bin/spark-shell）
 * Spark SQL的DataFrame API 允许我们使用 DataFrame 而不用必须去注册临时表或者生成SQL表达式。DataFrame API 既有
 * transformation操作也有action操作，DataFrame的转换从本质上来说更具有关系， 而 DataSet API 提供了更加函数式的 API
 *
 * 2.2.1	创建DataFrame
 * 在Spark SQL中SparkSession是创建DataFrame和执行SQL的入口，创建DataFrame有三种方式：
 *
 * ① 通过Spark的数据源进行创建；
 * ② 从一个存在的RDD进行转换；
 * ③ 还可以从Hive Table进行查询返回。
 *
 * 1)	从Spark数据源进行创建
 * 	查看Spark支持创建文件的数据源格式
 * scala> spark.read.
 * csv   format   jdbc   json   load   option   options   orc   parquet   schema   table   text   textFile
 *
 * 	读取json文件创建DataFrame
 * scala> val df = spark.read.json("/opt/module/spark-local/people.json")
 * df: org.apache.spark.sql.DataFrame = [age: bigint， name: string]
 *
 * 注意：如果从内存中获取数据，spark可以知道数据类型具体是什么，如果是数字，默认作为Int处理；
 * 但是从文件中读取的数字，不能确定是什么类型，所以用bigint接收，可以和Long类型转换，但是和Int不能进行转换
 *
 * 	展示结果
 * scala> df.show
 * +---+--------+
 * |age|    name|
 * +---+--------+
 * | 18|qiaofeng|
 * | 19|  duanyu|
 * | 20|    xuzhu|
 * +---+--------+
 *
 * 2)	从RDD进行转换
 * 2.5节我们专门讨论
 *
 * 3)	Hive Table进行查询返回
 * 3.4节我们专门讨论
 *
 *
 * 2.2.2	SQL风格语法
 * SQL语法风格是指我们查询数据的时候使用SQL语句来查询，这种风格的查询必须要有临时视图或者全局视图来辅助
 *
 * 1)	创建一个DataFrame
 * scala> val df = spark.read.json("/opt/module/spark-local/people.json")
 * df: org.apache.spark.sql.DataFrame = [age: bigint， name: string]
 *
 * 2)	对DataFrame创建一个临时表
 * scala> df.createOrReplaceTempView("people")
 *
 * 3)	通过SQL语句实现查询全表
 * scala> val sqlDF = spark.sql("SELECT * FROM people")
 * sqlDF: org.apache.spark.sql.DataFrame = [age: bigint， name: string]
 *
 * 4)	结果展示
 * scala> sqlDF.show
 * +---+--------+
 * |age|    name|
 * +---+--------+
 * | 18|qiaofeng|
 * | 19|  duanyu|
 * | 20|   xuzhu|
 * +---+--------+
 *
 * 注意：普通临时表是Session范围内的，如果想应用范围内有效，可以使用全局临时表。使用全局临时表时
 * 需要全路径访问，如：global_temp.people
 *
 * 5)	对于DataFrame创建一个全局表
 * scala> df.createGlobalTempView("people")
 *
 * 6)	通过SQL语句实现查询全表
 * scala> spark.sql("SELECT * FROM global_temp.people").show()
 * +---+--------+
 * |age|    name|
 * +---+--------+
 * | 18|qiaofeng|
 * | 19|  duanyu|
 * | 20|   xuzhu|
 * +---+--------+
 *
 * scala> spark.newSession().sql("SELECT * FROM global_temp.people").show()
 * +---+--------+
 * |age|    name|
 * +---+--------+
 * | 18|qiaofeng|
 * | 19|  duanyu|
 * | 20|   xuzhu|
 * +---+--------+
 *
 * 2.2.3	DSL风格语法
 * DataFrame提供一个特定领域语言(domain-specific language, DSL)去管理结构化的数据，
 * 可以在 Scala, Java, Python 和 R 中使用 DSL，使用 DSL 语法风格不必去创建临时视图了
 *
 * 1)	创建一个DataFrame
 * scala> val df = spark.read.json("/opt/module/spark-local /people.json")
 * df: org.apache.spark.sql.DataFrame = [age: bigint， name: string]
 *
 * 2)	查看DataFrame的Schema信息
 * scala> df.printSchema
 * root
 * |-- age: Long (nullable = true)
 * |-- name: string (nullable = true)
 *
 * 3)	只查看”name”列数据
 * scala> df.select("name").show()
 * +--------+
 * |    name|
 * +--------+
 * |qiaofeng|
 * |  duanyu|
 * |   xuzhu|
 * +--------+
 *
 * 4)	查看所有列
 * scala> df.select("*").show
 * +--------+---------+
 * |    name |age|
 * +--------+---------+
 * |qiaofeng|       18|
 * |  duanyu|       19|
 * |   xuzhu|       20|
 * +--------+---------+
 *
 * 5)	查看”name”列数据以及”age+1”数据
 * 注意:涉及到运算的时候, 每列都必须使用$
 * scala> df.select($"name",$"age" + 1).show
 * +--------+---------+
 * |    name|(age + 1)|
 * +--------+---------+
 * |qiaofeng|       19|
 * |  duanyu|       20|
 * |   xuzhu|       21|
 * +--------+---------+
 *
 * 6)	查看”age”大于”19”的数据
 * scala> df.filter($"age">19).show
 * +---+-----+
 * |age| name|
 * +---+-----+
 * | 20|xuzhu|
 * +---+-----+
 *
 * 7)	按照”age”分组，查看数据条数
 * scala> df.groupBy("age").count.show
 * +---+-----+
 * |age|count|
 * +---+-----+
 * | 19|    1|
 * | 18|    1|
 * | 20|    1|
 * +---+-----+
 *
 *
 * 2.2.4	RDD转换为DataFrame
 * 注意：如果需要RDD与DF或者DS之间操作，那么都需要引入 import spark.implicits._  （spark不是包名，
 * 而是sparkSession对象的名称，所以必须先创建SparkSession对象再导入. implicits是一个内部object）
 *
 * 前置条件
 * 	导入隐式转换并创建一个RDD
 * 	在/opt/module/spark-local/目录下准备people.txt
 * qiaofeng,18
 * xuzhu,19
 * duanyu,20
 *
 * scala> import spark.implicits._
 * import spark.implicits._
 *
 * scala> val peopleRDD = sc.textFile("/opt/module/spark-local/people.txt")
 * 输出
 * peopleRDD: org.apache.spark.rdd.RDD[String] = /opt/module/spark-local /people.txt MapPartitionsRDD[3] at textFile at <console>:27
 *
 * 1)	通过手动确定转换
 * scala> peopleRDD.map{x=> val fields=x.split(",");(fields(0),fields(1).trim.toInt)}.toDF("name","age").show
 * +--------+---+
 * |    name|age|
 * +--------+---+
 * |qiaofeng| 18|
 * |   xuzhu| 19|
 * |  duanyu| 20|
 * +--------+---+
 *
 * 2)	通过样例类反射转换（常用）
 * 	创建一个样例类
 * scala> case class People(name:String,age:Int)
 * 	根据样例类将RDD转换为DataFrame
 * scala> peopleRDD.map{x=> var fields=x.split(",");People(fields(0),fields(1).toInt)}.toDF.show
 * +--------+---+
 * |    name|age|
 * +--------+---+
 * |qiaofeng| 18|
 * |   xuzhu| 19|
 * |  duanyu| 20|
 * +--------+---+
 *
 * 3)	通过编程的方式（了解，一般编程直接操作RDD较少，操作hive或数据文件等较多）
 */
object Spark_SQL_2 {

  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkSession，该对象创建DataFrame和执行SQL的入口
    val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()

    val sc: SparkContext = spark.sparkContext

    val rdd: RDD[(String, Int)] = sc.makeRDD(Array(("lisi", 10), ("zs", 20), ("zhiling", 40)))

    // 映射出来一个RDD[Row]，因为DataFrame其实就是DataSet[Row]
    val rowRdd: RDD[Row] = rdd.map(x => Row(x._1, x._2))

    val structTypes: StructType = StructType(Array(StructField("name", StringType), StructField("age", IntegerType)))

    val dataFrame: DataFrame = spark.createDataFrame(rowRdd, structTypes)

    dataFrame.show()

    //4.关闭会话
    spark.stop()
  }
}

/**
 * 2.2.5	DataFrame转换为RDD
 *
 * 直接调用rdd即可
 * 1)	创建一个DataFrame
 * scala> val df = spark.read.json("/opt/module/spark-local/people.json")
 * df: org.apache.spark.sql.DataFrame = [age: bigint,name: string]
 *
 * 2)	将DataFrame转换为RDD注意：得到的RDD存储类型为Row
 * scala> val dfToRDD = df.rdd
 * dfToRDD: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[19] at rdd at <console>:29
 *
 * 3)	打印RDD
 * scala> dfToRDD.collect
 * res3: Array[org.apache.spark.sql.Row] = Array([18,qiaofeng], [19,duanyu], [20,xuzhu])
 *
 *
 * 2.3	DataSet
 * DataSet是具有强类型的数据集合，需要提供对应的类型信息。
 * 2.3.1	创建DataSet
 *
 * 1)	使用样例类序列创建DataSet
 * scala> case class Person(name: String, age: Long)
 * defined class Person
 *
 * scala> val caseClassDS = Seq(Person("wangyuyan",2)).toDS()
 *
 * caseClassDS: org.apache.spark.sql.Dataset[Person] = [name: string, age: Long]
 *
 * scala> caseClassDS.show
 * +---------+---+
 * |     name|age|
 * +---------+---+
 * |wangyuyan|  2|
 * +---------+---+
 *
 * 2)	使用基本类型的序列创建DataSet
 * scala> val ds = Seq(1,2,3,4,5,6).toDS
 * ds: org.apache.spark.sql.Dataset[Int] = [value: int]
 *
 * scala> ds.show
 * +-----+
 * |value|
 * +-----+
 * |    1|
 * |    2|
 * |    3|
 * |    4|
 * |    5|
 * |    6|
 * +-----+
 *
 * 注意:在实际使用的时候，很少用到把序列转换成DataSet，更多是通过RDD来得到DataSet
 *
 * 2.3.2	RDD转换为DataSet
 * SparkSQL能够自动将包含有样例类的RDD转换成DataSet，样例类定义了table的结构，样例类属性通过反射变成了表的列名。样例类可以包含诸如Seq或者Array等复杂的结构。
 *
 * 1)	创建一个RDD
 * scala> val peopleRDD = sc.textFile("/opt/module/spark-local/people.txt")
 *
 * peopleRDD: org.apache.spark.rdd.RDD[String] = /opt/module/spark-local/people.txt MapPartitionsRDD[19] at textFile at <console>:24
 *
 * 2)	创建一个样例类
 * scala> case class Person(name:String,age:Int)
 * defined class Person
 *
 * 3)	将RDD转化为DataSet
 * scala> peopleRDD.map(line => {val fields = line.split(",");Person(fields(0),fields(1). toInt)}).toDS
 *
 * res0: org.apache.spark.sql.Dataset[Person] = [name: string, age: Long]
 *
 * 2.3.3	DataSet转换为RDD
 * 调用rdd方法即可。
 *
 * 1)	创建一个DataSet
 * scala> val DS = Seq(Person("zhangcuishan", 32)).toDS()
 *
 * DS: org.apache.spark.sql.Dataset[Person] = [name: string, age: Long]
 *
 * 2)	将DataSet转换为RDD
 * scala> DS.rdd
 *
 * res1: org.apache.spark.rdd.RDD[Person] = MapPartitionsRDD[6] at rdd at <console>:28
 *
 * 2.4	DataFrame与DataSet的互操作
 *
 * 2.4.1	DataFrame转为DataSet
 * 1)	创建一个DateFrame
 * scala> val df = spark.read.json("/opt/module/spark-local/people.json")
 *
 * df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
 *
 * 2)	创建一个样例类
 * scala> case class Person(name: String,age: Long)
 * defined class Person
 *
 * 3)	将DataFrame转化为DataSet
 * scala> df.as[Person]
 *
 * res5: org.apache.spark.sql.Dataset[Person] = [age: bigint, name: string]
 * 这种方法就是在给出每一列的类型后，使用as方法，转成Dataset，这在数据类型是DataFrame又需要针对各个字段处理时极为方便。
 * 在使用一些特殊的操作时，一定要加上 import spark.implicits._ 不然toDF、toDS无法使用。
 *
 * 2.4.2	Dataset转为DataFrame
 *
 * 1)	创建一个样例类
 * scala> case class Person(name: String,age: Long)
 * defined class Person
 *
 * 2)	创建DataSet
 * scala> val ds = Seq(Person("zhangwuji",32)).toDS()
 *
 * ds: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]
 *
 * 3)	将DataSet转化为DataFrame
 * scala> var df = ds.toDF
 * df: org.apache.spark.sql.DataFrame = [name: string, age: bigint]
 *
 * 4)	展示
 * scala> df.show
 * +---------+---+
 * |     name|age|
 * +---------+---+
 * |zhangwuji| 32|
 * +---------+---+
 *
 * 2.5	RDD、DataFrame和DataSet之间的关系
 * 在SparkSQL中Spark为我们提供了两个新的抽象，分别是DataFrame和DataSet。他们和RDD有什么区别呢？首先从版本的产生上来看：
 * RDD (Spark1.0) —> Dataframe(Spark1.3) —> Dataset(Spark1.6)
 * 如果同样的数据都给到这三个数据结构，他们分别计算之后，都会给出相同的结果。不同是的他们的执行效率和执行方式。
 * 在后期的Spark版本中，DataSet有可能会逐步取代RDD和DataFrame成为唯一的API接口。
 *
 * 2.5.1	三者的共性
 * 1)	RDD、DataFrame、DataSet全都是spark平台下的分布式弹性数据集，为处理超大型数据提供便利;
 * 2)	三者都有惰性机制，在进行创建、转换，如map方法时，不会立即执行，只有在遇到Action如foreach时，三者才会开始遍历运算;
 * 3)	三者有许多共同的函数，如filter，排序等;
 * 4)	在对DataFrame和Dataset进行操作许多操作都需要这个包:import spark.implicits._（在创建好SparkSession对象后尽量直接导入）
 * 5)	三者都会根据 Spark 的内存情况自动缓存运算，这样即使数据量很大，也不用担心会内存溢出
 * 6)	三者都有partition的概念
 * 7)	DataFrame和Dataset均可使用模式匹配获取各个字段的值和类型
 *
 * 2.5.2	三者的区别
 *
 * 1)	RDD
 * 	RDD一般和Spark MLib同时使用
 * 	RDD不支SparkSQL操作
 *
 * 2)	DataFrame
 * 	与RDD和Dataset不同，DataFrame每一行的类型固定为Row，每一列的值没法直接访问，只有通过解析才能获取各个字段的值
 * 	DataFrame与DataSet一般不与 Spark MLib 同时使用
 * 	DataFrame与DataSet均支持 SparkSQL 的操作，比如select，groupby之类，还能注册临时表/视窗，进行 sql 语句操作
 * 	DataFrame与DataSet支持一些特别方便的保存方式，比如保存成csv，可以带上表头，这样每一列的字段名一目了然(后面专门讲解)
 *
 * 3)	DataSet
 * 	Dataset和DataFrame拥有完全相同的成员函数，区别只是每一行的数据类型不同。 DataFrame其实就是DataSet的一个特例
 * type DataFrame = Dataset[Row]
 *
 * 	DataFrame也可以叫Dataset[Row],每一行的类型是Row，不解析，每一行究竟有哪些字段，各个字段又是什么类型都无从得知，只能用上面提到的getAS方法或者共性中的第七条提到的模式匹配拿出特定字段。而Dataset中，每一行是什么类型是不一定的，在自定义了case class之后可以很自由的获得每一行的信息
 *
 * 2.5.3	三者的互相转化
 *
 *
 */
object Spark_SQL_2_1 {

}

